package com.hmall.search.autotoLoad;

import com.alibaba.fastjson.JSON;
import com.hmall.search.constant.ItemConstant;
import com.hmall.search.pojo.Item;
import com.hmall.search.pojo.ItemDoc;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
/**
 * @Auth: wan
 * @Desc: 多线程插入es数据
 * @Date: 2023/4/19 下午 9:31
 */
@Slf4j
public class ElasticsearchInsertThread extends Thread {

    private List<Item> items;
    private RestHighLevelClient client;

    public ElasticsearchInsertThread(List<Item> items, RestHighLevelClient client) {
        this.items = items;
        this.client = client;
    }

    @Override
    public void run() {
        try {
	        BulkResponse bulkResponse = null;
	        BulkRequest bulkRequest = new BulkRequest();
	        for (int i = 0; i < items.size(); i++) {
		        IndexRequest indexRequest = new IndexRequest(ItemConstant.INDEX_LIBRARY_NAME);
		        indexRequest
				        .id(items.get(i).getId().toString())
				        .source(JSON.toJSONString(items.get(i)), XContentType.JSON);
		        bulkRequest.add(indexRequest);

		        if (i % 1000 == 0) {
			        bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
			        bulkRequest = new BulkRequest();
		        }
	        }
	        if (bulkRequest.numberOfActions() > 0) {
		        bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
	        }
            if (bulkResponse.hasFailures()) {
                // 处理失败
	            log.error("信息处理失败！！！！");
            }
        } catch (Exception e) {
            // 处理异常
	        log.error(e.getMessage());
        }
    }
}

